package com.note.kafka.service;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * <p>【描述】：kafka 生产者</p>
 * <p>【作者】: BeyMax</p>
 * <p>【日期】: 2019-08-29</p>
 **/
@Service
public class KafkaProductServiceImpl implements KafkaProductService {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private KafkaTemplate kafkaTemplate;
    // 采集配置任务
//    @Value("${kafka.producer.topic1}")
    private String TASK_TOPIC;
    // 采集任务状态
//    @Value("${kafka.producer.topic2}")
    private String TASK_STATUS;
    // 帖子、博主、粉丝数据
//    @Value("${kafka.producer.topic3}")
    private String TASK_DATA;
    // 定时检测任务是否完成
//    @Value("${kafka.producer.topic4}")
    private String TASK_CHECK;
    // mysql同步数据最大id
//    @Value("${kafka.producer.topic5}")
    private String DATA_MAX_ID;
    //
//    @Value("${kafka.producer.topic6}")
    private String MAX_ID_BACK;

    /**
     * <p>【描述】：采集任务配置【生产者】</p>
     * <p>【方法】：sendTaskConfigKafka</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-29</p>
     */
    @Override
    public void sendTaskConfigKafka(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(TASK_TOPIC, "taskConfig", message);
            taskConfig.addCallback(o -> logger.info("采集任务配置【生产者】 send-消息发送成功：{}", message), throwable -> logger.info("采集任务配置【生产者】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("发送 采集任务配置【生产者】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：定时检测任务是否完成</p>
     * <p>【方法】：sendTaskStatusKafka</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-29</p>
     */
    @Override
    public void sendTaskStatusKafka(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(TASK_CHECK, "taskCheck", message);
            taskConfig.addCallback(o -> logger.info("【定时检测任务是否完成】 send-消息发送成功：{}", message), throwable -> logger.info("【定时检测任务是否完成】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【定时检测任务是否完成】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：采集已经完成的任务，需要更新mysql中采集任务配置表中的状态</p>
     * <p>【方法】：sendTaskFinishUpdateKafka</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-29</p>
     */
    @Override
    public void sendTaskFinishUpdateKafka(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(TASK_STATUS, "taskStatus", message);
            taskConfig.addCallback(o -> logger.info("【更新mysql中采集任务配置表中的状态】 send-消息发送成功：{}", message), throwable -> logger.info("【更新mysql中采集任务配置表中的状态】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【更新mysql中采集任务配置表中的状态】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：查询帖子正文已经同步到的最大id</p>
     * <p>【方法】：sendTransferTweetData</p>
     * <p>【参数】: [jo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-30</p>
     */
    @Override
    public void sendTransferTweetData(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(DATA_MAX_ID, "dataMaxId", message);
            taskConfig.addCallback(o -> logger.info("【查询帖子正文已经同步到的最大id】 send-消息发送成功：{}", message), throwable -> logger.info("【查询帖子正文已经同步到的最大id】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【查询帖子正文已经同步到的最大id】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：查询tweet用户信息表同步到的最大id</p>
     * <p>【方法】：sendTransferTweetUserData</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-30</p>
     */
    @Override
    public void sendTransferTweetUserData(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(DATA_MAX_ID, "userMaxId", message);
            taskConfig.addCallback(o -> logger.info("【查询tweet用户信息表同步到的最大id】 send-消息发送成功：{}", message), throwable -> logger.info("【查询tweet用户信息表同步到的最大id】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【查询tweet用户信息表同步到的最大id】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：tweet粉丝数据同步到的最大id</p>
     * <p>【方法】：sendTransferTweetUserFollow</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-30</p>
     */
    @Override
    public void sendTransferTweetUserFollow(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(DATA_MAX_ID, "followMaxId", message);
            taskConfig.addCallback(o -> logger.info("【tweet粉丝数据同步到的最大id】 send-消息发送成功：{}", message), throwable -> logger.info("【tweet粉丝数据同步到的最大id】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【tweet粉丝数据同步到的最大id】，kafka消息失败！", e);
        }
    }

    /**
     * <p>【描述】：postgresql已经同步的最大id，回写到mysql</p>
     * <p>【方法】：sendUpdateMaxIdBack</p>
     * <p>【参数】: [messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-30</p>
     */
    @Override
    public void sendUpdateMaxIdBack(JSONObject messageJo) {
        try {
            String message = messageJo.toJSONString();
            ListenableFuture taskConfig = kafkaTemplate.send(MAX_ID_BACK, "maxIdBack", message);
            taskConfig.addCallback(o -> logger.info("【postgresql已经同步的最大id】 send-消息发送成功：{}", message), throwable -> logger.info("【postgresql已经同步的最大id】 消息发送失败：{}", message));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("【postgresql已经同步的最大id】，kafka消息失败！", e);
        }

    }

    /**
     * <p>【描述】：发送tweet帖子、用户、粉丝数据</p>
     * <p>【方法】：sendTweetDataAllMessage</p>
     * <p>【参数】: [tableName, messageJo]</p>
     * <p>【作者】: BeyMax</p>
     * <p>【日期】: 2019-08-30</p>
     */
    @Override
    public void sendTweetDataAllMessage(String tableName, String message) {

        if (tableName.contains("blog_tweets")) {
            try {
                ListenableFuture taskConfig = kafkaTemplate.send(TASK_DATA, tableName, message);
                taskConfig.addCallback(o -> logger.info("【发送tweet帖子数据】 send-消息发送成功"), throwable -> logger.info("【发送tweet帖子数据】 消息发送失败!!!"));
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("【发送tweet帖子数据】，kafka消息失败！", e);
            }
        } else if (tableName.contains("blog_users")) {
            try {
                ListenableFuture taskConfig = kafkaTemplate.send(TASK_DATA, tableName, message);
                taskConfig.addCallback(o -> logger.info("【发送tweet用户数据】 send-消息发送成功"), throwable -> logger.info("【发送tweet用户数据】 消息发送失败!!!"));
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("【发送tweet用户数据】，kafka消息失败！", e);
            }
        } else if (tableName.contains("data_follow")) {

            try {
                ListenableFuture taskConfig = kafkaTemplate.send(TASK_DATA, tableName, message);
                taskConfig.addCallback(o -> logger.info("【发送tweet粉丝数据】 send-消息发送成功"), throwable -> logger.info("【发送tweet粉丝数据】 消息发送失败!!!"));
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("【发送tweet粉丝数据】，kafka消息失败！", e);
            }
        }else if (tableName.contains("tw_blogger")){
            try {
                ListenableFuture taskConfig = kafkaTemplate.send(TASK_DATA, tableName, message);
                taskConfig.addCallback(o -> logger.info("【发送任务表数据】 send-消息发送成功"), throwable -> logger.info("【发送任务表数据】 消息发送失败!!!"));
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("【发送任务表数据】，kafka消息失败！", e);
            }
        }
    }
}
